package com.jscloud.bigdata.flink.windows;

import org.apache.flink.api.common.functions.FlatMapFunction;
import org.apache.flink.api.common.functions.MapFunction;
import org.apache.flink.api.java.functions.KeySelector;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.windowing.assigners.TumblingProcessingTimeWindows;
import org.apache.flink.streaming.api.windowing.time.Time;
import org.apache.flink.util.Collector;

/**
 * * **需求：每隔5s时间，统计最近5s出现的数据**
 * * 代码实现：
 * *        时间间隔可以通过 Time.milliseconds(x)，Time.seconds(x)，Time.minutes(x)等其中的一个来指定。
 */
public class TumblingWindowByJava {

        public static void main(String[] args) throws Exception {
                StreamExecutionEnvironment environment = StreamExecutionEnvironment.getExecutionEnvironment();
                DataStreamSource<String> socketTextStream = environment.socketTextStream("bigdata01", 9999);
                Tuple2<String, Integer> tuple2 = new Tuple2<String, Integer>();
                socketTextStream.flatMap(new FlatMapFunction<String, String>() {
                                @Override
                                public void flatMap(String value, Collector<String> out) throws Exception {
                                        String[] words = value.split(" ");
                                        for (String word : words) {
                                                out.collect(word);
                                        }
                                }
                        }).map(new MapFunction<String, Tuple2<String, Integer>>() {
                                @Override
                                public Tuple2 map(String value) throws Exception {
                                        tuple2.setFields(value, 1);
                                        return tuple2;
                                }
                        }).keyBy(new KeySelector<Tuple2<String, Integer>, String>() {
                                @Override
                                public String getKey(Tuple2<String, Integer> value) throws Exception {
                                        return value.f0;
                                }
                        }).window(TumblingProcessingTimeWindows.of(Time.seconds(5)))
                        .sum(1)
                        .print();
                environment.execute();
        }
}